package mq

import (
	"github.com/streadway/amqp"
	"sync"
	"strconv"
	"fmt"
	"gitee.com/thomasgw/rabbitMQ-demo/message"
	"gitee.com/thomasgw/rabbitMQ-demo/client"
)

const (
	CALNAME = "calculator"
	KEY_ADD = "ADD"
	KEY_MINUS = "MIMUS"
)

type Exchanger struct {
	port      int
	ch        *amqp.Channel
	qmap      *sync.Map
	keySet    []string
	DataCh    chan message.Message
	cManager  *client.ClientManager
}

func NewExchanger(port int) *Exchanger {
	return  &Exchanger{
		port:     port,
		qmap:     new(sync.Map),
		keySet:   make([]string, 0),
		DataCh:   make(chan message.Message),
	}
}

func(exch *Exchanger) Subscribe(routingKeys []string) error {
	var (
		compKey string
		qs      []*amqp.Queue
	)
	for i,key := range routingKeys {
		if !exch.keyExist(key) {
			qname := key+"_"+"QUEUE"
			q,err := exch.ch.QueueDeclare(qname, false, false, true, false, nil)
			if err != nil {
				fmt.Println(err.Error())
				return err
			}
			exch.qmap.Store(key, &q)
			exch.keySet = append(exch.keySet, key)
			err = exch.ch.QueueBind(q.Name, key, CALNAME, false, nil)
			if err != nil {
				return err
			}
		}
		q,ok :=  exch.qmap.Load(key)
		if ok{
			qs = append(qs, q.(*amqp.Queue))
		}
		if i == len(routingKeys)-1 {
			compKey += key
		} else {
			compKey += (key + ".")
		}
	}
	fmt.Println(compKey)
	exch.cManager.RegisterClient(compKey, qs)
	return nil
}

func(exch *Exchanger) Start() error {
	conn,err := amqp.Dial("amqp://guest:guest@localhost:"+strconv.Itoa(exch.port)+"/")
	if err != nil {
		return err
	}
	exch.ch,err = conn.Channel()
	if err != nil {
		return err
	}
	exch.ch.ExchangeDeclare(
		CALNAME,       // name
		"direct",       // type
		true,          // durable
		false,         // auto-deleted
		false,         // internal
		false,         // no-wait
		nil,           // arguments
	)
	exch.cManager = client.NewClientManager(exch.ch)

	go func() {
		for event := range exch.DataCh {
			fmt.Println("Exchanger received event.")
			switch ev := event.(type){
			case *message.AddMessage:
				exch.notify(ev.Payload, KEY_ADD)
			case *message.MinusMessage:
				exch.notify(ev.Payload, KEY_MINUS)
			}
		}
	}()
	return nil
}

func(exch *Exchanger) notify(data []byte, keyType string) error {
	fmt.Println("Exchanger do 'Publish message.'")
	return exch.ch.Publish(
		CALNAME,
		keyType,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        data,
		})
}

func(exch *Exchanger) keyExist(key string) bool {
	for _,item := range exch.keySet {
		if item == key {
			return true
		}
	}
	return false
}

func(exch *Exchanger) GetChannel() *amqp.Channel {
	return exch.ch
}

func(exch *Exchanger) GetCManager() *client.ClientManager {
	return exch.cManager
}
